package code

import java.nio.file.{FileSystems, Paths}

import akka.NotUsed
import akka.stream.IOResult
import akka.stream.scaladsl.{FileIO, Flow, Framing, Keep, RunnableGraph}
import akka.util.ByteString

import scala.concurrent.Future

//电子书Page 93
//windows下运行定位资源好像有问题，linux下没问题

object BlueprintExample extends AkkaStreamsApp {
  val file = this.getClass.getClassLoader.getResource("current_inventory.csv")
  println(file.getPath) //这样写会在windows上打印以/开头的路径，不符合规范会报错Illegal char <:> at index 2
  println(file.toURI.toString)

  //这样返回的路径与平台无关，参见https://stackoverflow.com/questions/9834776/java-nio-file-path-issue
  println(Paths.get(file.toURI).toString)

  val inPath = FileSystems.getDefault.getPath(Paths.get(file.toURI).toString)
  //val inPath = FileSystems.getDefault.getPath(file.getPath)//不要这样写

  val outPath = FileSystems.getDefault.getPath("no_inventory.csv")
  val fileSource = FileIO.fromPath(inPath)
  val fileSink = FileIO.toPath(outPath)

  val csvHandler: Flow[String, List[String], NotUsed] =
    Flow[String]
      .drop(1)
      .map(_.split(",").toList)
      .log("csvHandler")

  //这个名称起得不太好，这是一种Flow吗？这不是一种Graph吗？
  val lowInventoryFlow: RunnableGraph[Future[IOResult]] =
    fileSource
      .via(Framing.delimiter(ByteString("\r\n"), Integer.MAX_VALUE))
      .map(_.utf8String)
      .log("Before CSV Handler")
      .via(csvHandler)
      .filter{list => list(2).toInt == 0}
      .log("After filter")
      .map { list =>
        ByteString(list.mkString(",")) ++ ByteString("\r\n")
      }.toMat(fileSink)(Keep.right)

  override def akkaStreamsExample: Future[_] =
    lowInventoryFlow.run()

  runExample
}